package demo.mqserver;


import demo.common.Consumer;
import demo.common.MqException;
import demo.mqserver.core.*;
import demo.mqserver.datacenter.DiskDataCenter;
import demo.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

//通过这个类，表示 虚拟主机
//每个虚拟主机下面都管理自己的 交换机，队列，绑定，消息 数据
//同时提供 API 供上层调用
@Slf4j
@Data
public class VirtualHost {

    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);


    //操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    //操作队列的锁对象
    private final Object queueLocker = new Object();



    public VirtualHost(String name){
        this.virtualHostName = name;

        //对于 MemoryDataCenter 来说，不需要进行额外的初始化，只需要把对象 new 出来即可
        //对于 DiskDataCenter 来说，需要进行初始化操作，建库建表和初始化数据
        diskDataCenter.init();

        //针对硬盘的数据，恢复到内存中
        try{
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            log.info("[VirtualHost] 恢复数据内存失败！");
        }
    }

    //创建交换机
    //如果交换机不存在就创建，如果存在就直接返回
    //返回值是 boolean，创建成功返回 true，创建失败返回 false
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments){
        //把虚拟主机的名字，加上虚拟主机作为前提
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                //1.判断交换机是否已经存在，直接通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange != null){
                    //该交换机已经存在
                    log.info("[VirtualHost] 交换机存在！exchangeName = " + exchangeName);
                    return true;
                }

                //2.真正创建交换机，构造 Exchange  对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                //3.把交换机对象写入硬盘中
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }

                //4.把交换机对象写入内存中
                memoryDataCenter.insertExchange(exchange);
                log.info("[VirtualHost] 交换机常见成功！exchangeName = " + exchangeName);

                //先写硬盘，再写内存  目的是硬盘更容易写失败，如果硬盘写失败了，内存就不写了
                //若先写内存，再写硬盘  硬盘写失败了，还需要把内存中的数据再删掉
            }
            return true;
        }catch (Exception e){
            log.error("[Virtual] 创建交换机失败！exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                //1.找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(toDelete == null){
                    throw new MqException("[VirtualHost] 交换机不存在，删除失败！exchangeName = " + exchangeName);
                }
                //2.删除硬盘上的数据
                if(toDelete.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //3.删除内存上的数据
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("[VirtualHost] 删除交换机成功！exchangeName = " + exchangeName);
            }
            return true;
        }catch (Exception e){
            log.error("[VirtualHost] 删除交换机失败！exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments){
        //把队列的名字拼接上虚拟主机的名字
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                //1.判断队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue != null){
                    log.info("[VirtualHost] 队列已经存在！queueName = " + queueName);
                    return true;
                }
                //2.创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setArguments(arguments);
                //3.写硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                //4.写内存
                memoryDataCenter.insertQueue(queue);
                log.info("[VirtualHost] 添加队列成功！ queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            log.error("[VirtualHost] 队列创建失败！ queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                //1.根据队列名，查询当前的队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName);
                }
                //2.删除硬盘数据
                if(queue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                //3.删除内盘数据
                memoryDataCenter.deleteQueue(queueName);
                log.info("[VirtualHost] 队列删除成功！queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            log.error("[VirtualHost] 队列删除失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.判断当前的绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existsBinding != null){
                        throw new MqException("[VirtualHost] 绑定已经存在！queueName = " + queueName + "，exchangeName = " + exchangeName);
                    }
                    //2.验证 bindingKey 是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] 绑定非法！bindingKey = " + bindingKey);
                    }
                    //3.创建绑定对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //4.获取对应的交换机和队列，如果不存在，绑定也就无法创建
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName = " + exchangeName);
                    }
                    //5.写硬盘
                    if(queue.isDurable() && exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    //6.写内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            log.info("[VirtualHost] 绑定创建成功！exchangeName = " + exchangeName + "，queueName = " + queueName);
            return true;
        }catch (Exception e){
            log.info("[VirtualHost] 绑定创建失败！exchangeName = " + exchangeName + "，queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueUnbind(String queueName, String exchangeName){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.获取绑定，检验绑定是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(binding == null){
                        throw new MqException("[VirtualHost] 删除绑定失败！exchangeName = " + exchangeName + "，queueName = " + queueName);
                    }
                    //2.无论绑定是否持久化，都尝试从硬盘删除（就算不存在，这个删除操作也无副作用）
                    diskDataCenter.deleteBinding(binding);
                    //3.删除内存上的数据
                    memoryDataCenter.deleteBinding(binding);
                    log.info("[VirtualHost] 删除绑定成功！");
                }
            }
            return true;
        }catch (Exception e){
            log.info("[VirtualHost] 删除绑定失败！");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
         try{
             //1.交换机的名字
             exchangeName = virtualHostName + exchangeName;
             //2.检查 routingKey 是否合法
             if(!router.checkRoutingKey(routingKey)){
                 throw new MqException("[VirtualHost] routingKey 非法！ routingKey = " + routingKey);
             }
             //3.查找交换机的类型
             Exchange exchange = memoryDataCenter.getExchange(exchangeName);
             if(exchange == null){
                 throw new MqException("[VirtualHost] 交换机不存在！exchangeName = " + exchangeName);
             }
             //4.判断交换机的类型
             if(exchange.getType() == ExchangeType.DIRECT){
                 //按照交换机的类型来转发消息
                 //以 routingKey 为队列的名字，直接把消息写入指定的队列中
                 //该类型可以无视绑定关系
                 String queueName = virtualHostName + routingKey;
                 //5.构造消息对象
                 Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                 //6.查找该队列名对应的对象
                 MSGQueue queue = memoryDataCenter.getQueue(queueName);
                 if(queue == null){
                     throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName);
                 }
                 //7.队列存在，直接给队列中写入消息
                 sendMessage(queue, message);
             }else{
                 //按照 fanout 和 topic 的方式来转发
                 //5.找到该交换机关联的所有绑定，并遍历这些绑定对象
                 ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                 for(Map.Entry<String, Binding> entry : bindingsMap.entrySet()){
                     //1）获取绑定对象，判断对应的队列是否存在
                     Binding binding = entry.getValue();
                     MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                     if(queue == null){
                         //此处不抛出异常，这里可能会有多个队列
                         //希望不要因为一个队列的失败，影响到其他队列的信息的传输
                         log.error("[VirtualHost] 发送消息时，发现队列不存在！queueName = " + binding.getQueueName());
                         continue;
                     }
                     //2）构造消息对象
                     Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                     //3）判断这个消息能否转发给该队列
                     //    如果是 fanout，所有绑定的队列都要转发
                     //    如果是 topic，判断 bindingKey 和 routingKey 是否匹配
                     if(!router.route(exchange.getType(), binding, message)){
                         continue;
                     }
                     //4）真正转发消息给队列
                     sendMessage(queue, message);
                 }

             }
             return true;
         }catch (Exception e){
             log.error("[VirtualHost] 消息发送失败！");
             e.printStackTrace();
             return false;
         }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //此处发送消息，就是把消息写入到 硬盘 和 内存 上
        int deliverMode = message.getDeliverMode();
        //deliverMode 为 1 不持久化，为 2 持久化
        if(deliverMode == 2){
            diskDataCenter.sendMessage(queue, message);
        }
        //写入内存
        memoryDataCenter.sendMessage(queue, message);
        //通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    //订阅消息
    //添加一个队列的订阅者，当队列收到消息后，把消息推送给对应的订阅者
    // consumerTag 消费者的身份标识
    // autoAck 消息被消费后的应答方式  true 为自动应答，false 为手动应答
    // consumer 是一个回调函数，此处类型设置为函数式接口，后续调用 basicConsume 并且传实参时，就可以写作 lambda 这样的
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        //构造一个 ConsumerEnv 对象，找到对应的队列，再把这个 Consumer 对象添加到该队列中
        queueName = virtualHostName + queueName;
        try{
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            log.info("[VirtualHost] basicConsume 成功！queueName = " + queueName);
            return true;
        }catch (Exception e){
            log.error("[VirtualHost] basicConsume 失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try{
            //1.获取到消息队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在！messageId = " + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("[VirtualHost] 要确认的队列不存在！queueName = " + queueName);
            }
            //2.删除硬盘上的数据
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue, message);
            }
            //3.删除消息中心中的数据
            memoryDataCenter.removeMessage(messageId);
            //4.删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            log.info("[VirtualHost] basicAck 成功，消息确认成功！queueName = " + queueName + "，messageId = " + messageId);
            return true;
        }catch (Exception e){
            log.error("[VirtualHost] basicAck 成功，消息确认失败！queueName = " + queueName + "，messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
